package cn.caplike.demo.kafka.producer.transaction;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

public class ProducerTransaction {

    public static void main(String[] args) {
        // 在控制台监听一个 topic: ./kafka-console-consumer.sh --bootstrap-server big-data-1:9092 big-data-2:9092 big-data-3:9092 --topic first

        // 配置
        final Properties properties = new Properties();

        // 连接集群 bootstrap.servers
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "big-data-1:9092,big-data-2:9092,big-data-3:9092");
        // 指定对应 key 和 value 的序列化类型 key.serializer
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        // 指定事务 ID
        properties.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "transaction_id#A");

        // 创建生产者对象
        try (final KafkaProducer<String, String/* "", hello */> producer = new KafkaProducer<>(properties)) {
            // 必须指定全局唯一的 transaction.id
            producer.initTransactions();
            producer.beginTransaction();

            try {
                // 普通异步发送数据
                producer.send(new ProducerRecord<>("first", "caplike")); // 会自动创建 topic, 发送一条消息, 所有消费者都能收到

                int i = 1 / 0; // 消息不会发送

                producer.commitTransaction();
            } catch (Exception e) {
                producer.abortTransaction();
                throw new RuntimeException(e);
            }
        }

    }
}
